package com.bw.gmall.realtime.app.dwd;

import com.bw.gmall.realtime.utils.MyKafkaUtil;
import com.bw.gmall.realtime.utils.MysqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdInteractionComment {
    public static void main(String[] args) {
        //todo 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // TODO 3. 从 Kafka 读取业务数据，
        tenv.executeSql("create table topic_db(" +
                "`database` string, " +
                "`table` string, " +
                "`type` string, " +
                "`data` map<string,string>, " +
                "`proc_time` as proctime(), " +
                "`ts` string " +
                ")"+ MyKafkaUtil.getKafkaDDL("topic_db","dwd_interaction_comment"));

        // TODO 4. 读取评论表数据
        Table commentInfo = tenv.sqlQuery("select " +
                "data['id'] id, " +
                "data['user_id'] user_id, " +
                "data['sku_id'] sku_id, " +
                "data['order_id'] order_id, " +
                "data['create_time'] create_time, " +
                "data['appraise'] appraise, " +
                "proc_time, " +
                "ts " +
                "from topic_db " +
                "where `table` = 'comment_info' " +
                "and `type` = 'insert' ");
        tenv.createTemporaryView("comment_info",commentInfo);


        // TODO 5. 建立 MySQL-LookUp 字典表

        tenv.executeSql(MysqlUtil.getBaseDicLookUpDDL());


        // TODO 6. 关联两张表
        Table resTable = tenv.sqlQuery("select " +
                "ci.id, " +
                "ci.user_id, " +
                "ci.sku_id, " +
                "ci.order_id, " +
                "date_format(ci.create_time,'yyyy-MM-dd') date_id, " +
                "ci.create_time, " +
                "ci.appraise, " +
                "dic.dic_name, " +
                "ts " +
                "from comment_info ci " +
                "join " +
                "base_dic for system_time as of ci.proc_time as dic " +
                "on ci.appraise = dic.dic_code ");
        tenv.createTemporaryView("res_table",resTable);

        // TODO 7. 建立 Kafka-Connector dwd_interaction_comment 表

        tenv.executeSql("create table dwd_interaction_comment (" +
                "id string, " +
                "user_id string, " +
                "sku_id string, " +
                "order_id string, " +
                "date_id string, " +
                "create_time string, " +
                "appraise_code string, " +
                "appraise_name string, " +
                "ts string " +
                ")"+MyKafkaUtil.getKafkaSinkDDL("dwd_interaction_comment"));


        // TODO 8. 将关联结果写入 Kafka-Connector 表
        tenv.executeSql("insert into dwd_interaction_comment select * from res_table");
    }
}
